#include "app_pool.h"
#include "mqueue.h"
#include "pthread.h"
#include <stdio.h> // 补充缺失的头文件

// 定义每个消息队列中的任务结构体,包含执行函数和参数
typedef struct
{
    void *(*func)(void *); // 指向执行函数的指针
    void *arg;             // 参数
} task_t;

#define MAX_THREAD_NUM 10
#define POOL_QUEUE_NAME "/gateway_pool_queue" // 消息队列名称

static pthread_t threads[MAX_THREAD_NUM];

static mqd_t mq;

/**
 *@brief:线程池的工作内容
 定义线程要执行什么工作,每个线程启动一个死循环,不停地尝试从队列中取任务
    每个任务包含执行函数和参数
    每个线程要取出函数,用函数执行参数
 */
static void *app_pool_thread_func(void *arg)
{
    while (1)
    {
        task_t task;
        // 从消息队列中取出任务
        ssize_t task_size = mq_receive(mq, (char *)&task, sizeof(task), NULL);
        if (task_size > 0)
        {
            // 执行任务
            task.func(task.arg);
        }
    }
    // 线程函数（如配合 pthread_create 使用）若声明为 void* 返回值类型，必须显式返回 NULL。
    return NULL;
}

/**
 * @brief 初始化线程池,创建并开辟线程数组,创建消息队列
 *
 * @return 0 成功，-1 失败
 */
int app_pool_init()
{
    // 创建开辟线程数组
    for (int i = 0; i < MAX_THREAD_NUM; i++)
    {
        // 每个线程都要执行
        pthread_create(&threads[i], NULL, app_pool_thread_func, NULL);
        printf("创建线程成功: %d\n", i);
    }
    // 清空历史队列
    mq_unlink(POOL_QUEUE_NAME);
    // 设置消息队列属性
    struct mq_attr attr;
    attr.mq_maxmsg = 100;
    attr.mq_msgsize = sizeof(task_t);
    // 创建消息队列
    mq = mq_open(POOL_QUEUE_NAME, O_RDWR | O_CREAT, 0666, &attr);
    if (mq == -1)
    {
        perror("mq_open, error"); // 打印错误信息(此函数会自动补充原因)
        return -1;
    }
    return 0;
}

/**
 *@brief:对外提供一个函数,用来向线程池中提交任务
 */
int app_pool_submit_task(void *(*func)(void *), void *arg)
{
    // 把函数和参数组合成一个任务结构体
    task_t task;
    task.func = func;
    task.arg = arg;
    // 把任务结构体放入消息队列中:把任务结构体当成了数据,传到消息队列中
    int ret = mq_send(mq, (char *)&task, sizeof(task), 0);
    if (ret == -1)
    {
        perror("mq_send, error");
        return -1;
    }
    return 0;
}

/**
 * @brief 销毁线程池,释放线程数组,销毁消息队列
 *
 * @return 0 成功，-1 失败
 */
void app_pool_destroy()
{
    // 销毁线程池
    for (int i = 0; i < MAX_THREAD_NUM; i++)
    {
        pthread_cancel(threads[i]);
    }
    // 销毁消息队列
    mq_close(mq);
    mq_unlink(POOL_QUEUE_NAME);
}
